package org.smile.io.buff;


import org.smile.log.LoggerHandler;

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;

/**
 * 管道实现的缓冲区
 */
public class PipedBuffer extends AbstractByteBuff implements LoggerHandler {

    /**
     * 管道读取方
     */
    PipedInputStream reader;
    /***
     * 管道写入方
     */
    PipedOutputStream writer;

    Thread readTaskThread;

    int pipedSize;

    boolean running=true;

    public PipedBuffer(int size){
        this.pipedSize=size;
        this.reader=new PipedInputStream(size);
        try {
            this.writer=new PipedOutputStream(reader);
        } catch (IOException e) {
           throw new RuntimeException(e);
        }
    }

    @Override
    public int read() throws IOException {
        return reader.read();
    }

    @Override
    public long skip(long n) throws IOException {
        return reader.skip(n);
    }

    @Override
    public int readabled() throws IOException {
        return reader.available();
    }

    @Override
    public void mark() throws IOException {
        this.reader.mark(1024);
    }

    @Override
    public void reset() throws IOException {
        this.reader.reset();
    }

    @Override
    public int readBytes(byte[] bytes) throws IOException {
        return this.reader.read(bytes);
    }

    @Override
    public int writeabled() throws IOException {
        return this.pipedSize-this.reader.available();
    }

    @Override
    public byte[] toByteArray() throws IOException {
       int size=this.reader.available();
       byte[] bytes=new byte[size];
       this.reader.read(bytes);
       return bytes;
    }

    @Override
    public void clear() throws IOException {
        throw new IOException("not support");
    }

    @Override
    public void write(int b) throws IOException {
        this.writer.write(b);
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
        this.writer.write(b, off, len);
    }

    @Override
    public int read(byte[] bytes, int offset, int len) throws IOException {
        return this.reader.read(bytes, offset, len);
    }

    /**
     * 启动读取任务
     * @param task
     */
    public void startReadTask(final ReadTask task){
        if(this.readTaskThread!=null){
            throw new RuntimeException("read task is exists");
        }
        this.readTaskThread=new Thread(){
            @Override
            public void run() {
                while(running) {
                    try {
                        if(PipedBuffer.this.readabled()>0){
                            task.run(PipedBuffer.this);
                        }
                    }catch(Exception e){
                       logger.error(e);
                    }
                }
            }
        };
        readTaskThread.start();
    }

    public void pause (boolean pause){
        this.running =!pause;
    }

    public static interface ReadTask {
        public void run(PipedBuffer buffer);
    }
}
